perf(pipeline): defer only for shards the batch has pending work on (#513) - #708

…513)

`must_wait_for_pending_remote`'s multi-key arm answered "wait" on the command
NAME, without asking where its keys were. But `remote_groups` only ever holds
FOREIGN shards -- the slotting branch is `else if let Some(target) =
target_shard`, and `target_shard` is `None` for a local key -- so the moon#507
hazard (reading state a pending command is about to write, or writing state it
then overwrites) requires the two to meet ON THE SAME SHARD. An MGET reading
shards the batch has no pending work for was being cut for nothing.

The cut is not free: it ends the batch pass, and the phase-2b drain then
dispatches one PipelineBatchSlotted per target shard and awaits each reply slot
in turn.

The guard now takes a `pending` bitmask, maintained beside `remote_groups` at
O(1) (set on insert, cleared with the map), and compares it against the shard
mask of the command's keys. The mask comes from the shared key-position walker
(moon#582) -- the same one ACL, cache invalidation and
`cross_shard_multikey_rejection` use -- so layouts like ZUNIONSTORE are
enumerated by the code that already knows them rather than a second copy.

Only the multi-key arm is refined; the other two still always wait, because
neither can be bounded by a key mask. An inline-intercepted command (EVAL,
SWAPDB) executes against the LOCAL slice whatever keys it declares, and a
keyless command (FLUSHALL, KEYS, SCAN) touches every shard. Every case the
mask cannot enumerate -- `SORT ... BY w_*`, a key position holding a
non-string, more shards than the mask has bits -- also waits: wrongly waiting
costs a batch boundary, wrongly proceeding corrupts data.

The two predicates were folded into one. The unmasked form had no callers left
and keeping it would have left two answers to the same question.

Measured on moon-dev (aarch64, 6 vCPU), --shards 4, 32 interleavings of
SET,SET,MGET, six fresh server starts per side, interleaved:

  MGET reads shards the writes never touch: 41,600 -> 86,500 ops/s (2.08x),
                                            64 deferrals -> 0
  MGET reads the shards being written:      34,300 ops/s, 64 deferrals, both

Fresh starts per measurement because SO_REUSEPORT decides which shard the
connection lands on, and that changes the shape's cost as much as the code
does. The deferral counts are placement-independent: 64/64 before and 0/0
after in every round.

A co-located {tag} multi-key command still defers, and must: the coordinator
executes it inline rather than slotting it, so skipping the wait would re-open
moon#507. Routing a single-owner multi-key command into the slotted batch is
tracked separately.

Tests: `pco13` drives the disjoint shape and asserts 0 deferrals, with an
overlap leg on the same harness that must stay non-zero -- without it a green
disjoint leg could just mean the writes never went cross-shard. Both legs pin
ONE key per shard rather than "the first n keys in this set", because two keys
that both hashed to shard 0 made the overlap leg depend on where SO_REUSEPORT
put the connection (caught as a real flake while benchmarking). Four unit tests
cover the mask itself, including every fail-closed path. Verified by mutation:
reverting the guard to name-only fails pco13 alone; making it never wait fails
five of the moon#507 correctness tests.

Refs #513, #507, #512
author: Tin Dang

…actually use

The shard mask added earlier in this branch hashes the keys visible AT the
guard. In a workspace connection those are the RAW keys:
`workspace_rewrite_args` rebinds `cmd_args` further down the batch loop, and
the guard cannot move below it -- the connection-level intercepts it exists to
hold back (AUTH, CLIENT, CONFIG, INFO, SELECT, ...) run in between.

The discrepancy is not small. A workspace key is `{<32-hex>}:<key>`, and that
prefix is a hash TAG, so every key in a workspace routes to ONE shard however
the raw names scatter. A mask read off raw names therefore calls a command
disjoint from the very shard its own batch's writes are pending on -- moon#507
reopened for exactly the connections that opted into isolation.

Measured on the pre-fix build of this branch: 5 of 12 workspace connections
had `SET a; SET b; MGET a b` answer `$-1 $-1` for keys the same batch had
already acked `+OK`. The same test is green on the commit this branch forked
from, so this was introduced here, not uncovered here.

Fix: treat every shard as pending when the connection has a workspace, which
makes `must_wait_for_pending_remote` answer exactly as it did before the mask
existed. Workspace connections lose the batch-cut saving; they keep their
data.

Tests: `pco14` drives 12 workspace connections and asserts the MGET observes
its own batch's writes. It asserts CORRECTNESS rather than a deferral count
because the count is only wrong when the workspace's shard is foreign to the
connection and SO_REUSEPORT decides that -- the correctness claim holds for
every connection, so twelve make placement moot. Run 8x consecutively, green.
Verified by mutation: dropping the workspace arm reproduces 5/12 losses.

`pco13` gains a single-shard leg -- the shape the mask is most tempted to wave
through -- and its final correctness block is relabelled: those keys span two
shards, so calling them "co-located" was wrong.

Refs #513, #507, #702
author: Tin Dang